
package com.lee.fpflink; 
 
import com.lee.enumeration.FPFlinkQualifierEnum; 
import com.lee.info.CkFlowInfo; 
import com.lee.logger.Logger; 
import com.lee.tree.FPStreamGenerate; 
import com.lee.tree.FrequentFound; 
import com.lee.tree.PatternTree; 
import com.lee.tree.TwoTuple; 
import com.lee.utils.DMapBlackUtils; 
import com.lee.utils.DMapUtils; 
import org.apache.flink.streaming.api.functions.windowing.WindowFunction; 
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow; 
import org.apache.flink.util.Collector; 
 
import java.io.IOException; 
import java.util.ArrayList; 
import java.util.Arrays; 
import java.util.Iterator; 
import java.util.List; 
import java.util.stream.Collectors; 
 
public class FindFrequentWindowFunction implements WindowFunction<CkFlowInfo, List<TwoTuple<List<String>, Long>>, String, GlobalWindow> { 
 
    @Override
    public void apply(String key, GlobalWindow globalWindow, Iterable<CkFlowInfo> input, Collector<List<TwoTuple<List<String>, Long>>> collector) throws Exception {
        FrequentFound frequentFound = new FrequentFound();
        Iterator<CkFlowInfo> iterator = input.iterator();
        List<CkFlowInfo> datasetCkFlow = new ArrayList<>();
        iterator.forEachRemaining(item->{
                datasetCkFlow.add(item);

        });
        Logger.info(key,"FindFrequentWindowFunction -> 过滤后第一条数据",datasetCkFlow.get(0).getCkList());
        Logger.info(key,"FindFrequentWindowFunction -> 过滤后第一条数据数据量",datasetCkFlow.get(0).getCkList().size());
        long index = DMapUtils.getItemIndex(FPFlinkQualifierEnum.DCOUNT.getCode());
        long dataIndex = datasetCkFlow.size();
 
        try {
            Logger.info(key,"-",index,"FindFrequentWindowFunction -> 找到频繁集");
            List<TwoTuple<List<String>, Long>> frequentItemsets = frequentFound.findFrequentItemsetsCkflow(datasetCkFlow);
            collector.collect(frequentItemsets);
 
        }catch (Exception e ){
            e.printStackTrace();
            Logger.error(key,"-",index,"FindFrequentWindowFunction -> 报错",e);
        }
    }
} 
